#include "multicaster.h"

handler::multicaster::multicaster(boost::shared_ptr<network::message_queue> q, 
					int num_threads)
: _message_q(q)
{
	boost::shared_ptr<handler::event_worker> worker;
	for(int i = 0; i < num_threads; ++i) {
		worker.reset(new handler::event_worker(_this, _message_q));
		worker->start_thread();
		_workers.push_back(worker);
	}
}

handler::multicaster::~multicaster()
{
	_message_q->interrupt();
	typedef list<boost::shared_ptr<handler::event_worker> >::iterator it;
	for(it i = _workers.begin(); i != _workers.end(); ++i) {
		(*i)->interrupt();
		(*i)->join();
		(*i).reset();
	}
	_workers.clear();
	_handler.clear();
}

void handler::multicaster::dispatch(boost::shared_ptr<network::message> msg)
{
	list<handler::multicaster::event_storage> copy;
	{
		boost::mutex::scoped_lock l(_monitor);
		copy = _handler;
	}
	typedef list<handler::multicaster::event_storage>::iterator it;
	for(it i = copy.begin(); i != copy.end(); ++i) {
		if(i->type == msg->type && i->socket == msg->sock) {
			i->handler->handle_event(msg);
		}
	}
}

void handler::multicaster::add_handler(network::event_type t, 
	boost::shared_ptr<network::i_socket> sock, 
	boost::shared_ptr<handler::event_handler> h) 
{
	h->add_listener(boost::shared_ptr<handler::multicaster>(this, 
							core::null_deleter()));
	boost::mutex::scoped_lock l(_monitor);
	handler::multicaster::event_storage store;
	store.type = t;
	store.socket = sock;
	store.handler = h;
	_handler.push_back(store);
}

void handler::multicaster::remove_handler(
	network::event_type t,
	boost::shared_ptr<network::i_socket> sock,
	boost::shared_ptr<handler::event_handler> h) 
{
	boost::mutex::scoped_lock l(_monitor);
	typedef list<handler::multicaster::event_storage>::iterator it;

	it i = _handler.begin();
	it next = _handler.begin();
	while(i != _handler.end()) {
		++next;
		if(i->handler == h && i->type == t && i->socket == sock) {
			_handler.erase(i);
		}
		i = next;
	}
}

void handler::multicaster::notify(const network::notifier *n)
{
	const handler::event_handler *h = 
		dynamic_cast<const handler::event_handler*>(n);
	if(!h) {
		return;
	}
	boost::mutex::scoped_lock l(_monitor);
	typedef list<handler::multicaster::event_storage>::iterator it;

	it i = _handler.begin();
	it next = _handler.begin();
	while(i != _handler.end()) {
		++next;
		if(i->handler.get() == h) {
			_handler.erase(i);
		}
		i = next;
	}
}

boost::shared_ptr<handler::multicaster> handler::multicaster::create( 
	boost::shared_ptr<network::message_queue> q, int num_threads)
{
	boost::shared_ptr<handler::multicaster> ptr(
		new handler::multicaster(q, num_threads));
	ptr->_this = ptr;
	typedef list<boost::shared_ptr<handler:: event_worker> >::iterator it;
	for(it i = ptr->_workers.begin(); i != ptr->_workers.end(); ++i) {
		(*i)->set_multicaster(ptr->_this);
	}
	return ptr;
}
	
